package myapps6;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * kafka 事务 demo
 */
public class ProducerTransactionSend {
	public static final String topic = "heima";
	public static final String brokerList = "192.168.0.150:9092";
	public static final String transactionId = "transactionId";

	public static void main(String[] args) {
		Properties properties = new Properties();
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);

		//transaction config
		properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionId);
		properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);

		//关闭自动提交
//		properties.put(ProducerConfig.C)

		KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

		//初始化事务
		producer.initTransactions();
		//开启事务
		producer.beginTransaction();
		try{
			//处理业务逻辑
			ProducerRecord<String,String> record = new ProducerRecord<>(topic,"message-1");
			producer.send(record);

			ProducerRecord<String,String> record1 = new ProducerRecord<>(topic,"message-2");
			producer.send(record1);
			//模拟异常
			// int i = 1/0;
			ProducerRecord<String,String> record2 = new ProducerRecord<>(topic,"message-3");
			producer.send(record2);

			producer.commitTransaction();
		}catch (Exception e){
			producer.abortTransaction();
			e.printStackTrace();
		}

		//
		producer.close();

	}

}
